package com.alibaba.datax.plugin.writer.spdhbasewriter;

import com.alibaba.datax.common.base.IdRecord;
import com.alibaba.datax.common.constant.IdRecordConstants;
import com.alibaba.datax.common.element.Record;
import com.alibaba.datax.common.plugin.RecordReceiver;
import com.alibaba.datax.common.spi.Writer;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.common.util.RedissonUtils;
import com.alibaba.fastjson2.JSONObject;
import org.redisson.api.RTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

//TODO writeProxy
public class SpdHbaseWriter extends Writer {


    public static class Job extends Writer.Job {
        private Configuration originalConfig = null;


        @Override
        public void preCheck(){
            this.init();
        }

        @Override
        public void init(){
            this.originalConfig = super.getPluginJobConf();

        }

        // 一般来说，是需要推迟到 task 中进行pre 的执行（单表情况例外）
        @Override
        public void prepare() {
        }

        @Override
        public List<Configuration> split(int mandatoryNumber) {
            List<Configuration> configurations = new ArrayList<>();
            configurations.add(this.originalConfig);
            return configurations;
        }

        // 一般来说，是需要推迟到 task 中进行post 的执行（单表情况例外）
        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }

    public static class Task extends Writer.Task {
        private static final Logger LOG = LoggerFactory
                .getLogger(Task.class);
        private Configuration writerSliceConfig;
        Connection conn;

        private String insertSql;
        @Override
        public void init() {
            this.writerSliceConfig = super.getPluginJobConf();
            insertSql = writerSliceConfig.getString("insertSql");
            String className = writerSliceConfig.getString("className");
            String url = writerSliceConfig.getString("url");

            try{
                //className = "org.apache.phoenix.jdbc.PhoenixDriver"
                Class.forName(className);
                Properties properties = new Properties();
                properties.setProperty("phoenix.schema.isNamespaceMappingEnabled","true");
                properties.setProperty("phoenix.schema.mapSystemTablesToNamespace","true");
                //url = "jdbc:phoenix:10.17.65.85:2181"
                conn = DriverManager.getConnection(url,properties);

            }catch (Exception e){
                e.printStackTrace();
            }

        }

        @Override
        public void prepare() {
        }

        public void startWrite(RecordReceiver recordReceiver) {

            String redisHost = System.getProperty("redisHost");
            String redisPasswd = System.getProperty("redisPasswd");
            String redisDatabase = System.getProperty("redisDatabase");
            String taskId = System.getProperty("taskId");
            String incrementField = writerSliceConfig.getString("incrementField");
            String tableName = writerSliceConfig.getString("tableName");

            LOG.info("开始写入数据任务:{}",taskId);
            RedissonUtils redissonUtils = new RedissonUtils(redisHost, redisPasswd, 6379, Integer.parseInt(redisDatabase));
            RTopic idRecord = redissonUtils.getRedissonClient().getTopic("idRecord");

            Record record;
            try {
                int count = 0;
                PreparedStatement ps = conn.prepareStatement(insertSql);
                String lastVal = "";
                while ((record = recordReceiver.getFromReader()) != null) {
                    Map<String, String> meta = record.getMeta();
                    lastVal = meta.get(incrementField);

                    String id = record.getColumn(0).asString();
                    String querySql = String.format("select 1 from %s where id = '%s'",tableName,id);
                    ResultSet resultSet = conn.createStatement().executeQuery(querySql);
                    if(resultSet.next()){
                        break;
                    }

                    for (int i=0;i<record.getColumnNumber();i++){
                        ps.setString(i+1,record.getColumn(i).asString());
                    }
                    ps.addBatch();
                    count++;
                    if(count>1000){
                        ps.executeBatch();
                        conn.commit();
                        idRecord.publish(JSONObject.toJSONString(new IdRecord(taskId,lastVal, IdRecordConstants.WRITE)));
                        ps.clearBatch();
                        count = 0;
                    }
                }
                ps.executeBatch();
                conn.commit();
                idRecord.publish(JSONObject.toJSONString(new IdRecord(taskId,lastVal, IdRecordConstants.WRITE)));
            }catch (Exception e){
               e.printStackTrace();
            }
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

    }


}
